package cn.zn.modules.kettle.service.impl;

import cn.zn.framework.common.exception.RRException;
import cn.zn.framework.common.validator.Assert;
import cn.zn.modules.kettle.dto.KettleMongo;
import cn.zn.modules.kettle.dto.KettleMysql;
import cn.zn.modules.kettle.dto.KettleParams;
import cn.zn.modules.kettle.service.KettleJobService;
import com.mongodb.client.result.DeleteResult;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.plugins.PluginFolder;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.mongo.MongoProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import java.util.Set;

@Service
@Slf4j
@ConfigurationProperties("spring.datasource.druid")
public class KettleJobServiceImpl implements KettleJobService {
    @Autowired
    KettleMysql kettleMysql;
    @Autowired
    KettleMongo kettleMongo;
    @Autowired
    KettleParams kettleParams;
    @Autowired
    private MongoProperties mongoParameters;
    @Autowired
    private MongoTemplate mongoTemplate;


    @Override
    public void execJobTurn(Integer ap, String mysqlDbName) {
        Assert.isNull(ap, "ap 不能为空");
        Assert.isBlank(mysqlDbName, "mysqlDbName 不能为空");
        try {
            // 设置插件路径
            StepPluginType.getInstance().getPluginFolders().add(new PluginFolder(kettleParams.getPluginPath(), false, true));
            KettleEnvironment.init();
            // 设置job路径
            JobMeta jobMeta = new JobMeta(kettleParams.getJobPath(), null);
            Job job = new Job(null, jobMeta);
            job.setVariable("ap_current", String.valueOf(ap));
            job.setVariable("mongo_ip", mongoParameters.getHost());
            job.setVariable("mongo_port", String.valueOf(mongoParameters.getPort()));
            job.setVariable("mongo_authdb", mongoParameters.getAuthenticationDatabase());
            job.setVariable("mongo_username", mongoParameters.getUsername());
            job.setVariable("mongo_password", String.valueOf(mongoParameters.getPassword()));
//            MysqlParam mysqlParam = this.parseMysqlUrl(url);
            job.setVariable("mysql_ip", kettleMysql.getIp());
            job.setVariable("mysql_port", kettleMysql.getPort());
            job.setVariable("mysql_db", mysqlDbName.concat("?allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT"));
            job.setVariable("mysql_username", kettleMysql.getUsername());
            job.setVariable("mysql_password", kettleMysql.getPassword());
            job.start();
            job.waitUntilFinished();
            if (job.getErrors() > 0) {
                throw new RRException("数据抽取到mongo时出错");
            }
        } catch (Exception e) {
            log.error("", e);
            throw new RRException("数据抽取到mongo时出错");
        }
    }

    @Override
    public void cleanByAp(Integer ap) {
        Set<String> collectionNames = mongoTemplate.getCollectionNames();
        collectionNames.stream().forEach(o -> {
            DeleteResult deleteResult = mongoTemplate.remove(new Query().addCriteria(Criteria.where("ap").is(ap)), o);
            log.debug("{}", deleteResult.getDeletedCount());
        });
    }

    private MysqlParam parseMysqlUrl(String url) {
        String str = url.substring(url.lastIndexOf("//") + 2, url.indexOf("?"));
        String[] strArr = str.split("/");
        String[] ipPort = strArr[0].split(":");
        return MysqlParam.builder()
                .database(strArr[1])
                .ip(ipPort[0])
                .port(ipPort[1])
                .build();
    }
}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
class MysqlParam{
    private String database;
    private String ip;
    private String port;
}
